package com.shujia.filter;

import com.shujia.utils.HbaseTool;
import org.apache.hadoop.hbase.CompareOperator;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.hbase.filter.*;
import org.apache.hadoop.hbase.util.Bytes;

/**
 * 三大类过滤器：
 * 1. 比较过滤器
 * 2. 专用过滤器
 * 3. 包装过滤器
 * <p>
 * 专用过滤器：
 * 通过SingleColumnValueFilter与查询文科班所有学生信息
 */
public class HBaseZhuanYongFilter {
    public static void main(String[] args) throws Exception {
        // 通过SingleColumnValueFilter与查询文科班所有学生信息
//        xuQiu1();
        // 通过SingleColumnValueExcludeFilter与BinaryComparator查询文科一班所有学生信息，最终不返回clazz列
//        xuQiu2();
        //rowkey前缀过滤器
        // 过滤出行键以15001005开头
        //做法1：使用比较过滤器实现，行键过滤器+前缀比较器
//        xuQiu3();
        //做法2：专用过滤器，PrefixFilter
//        xuQiu4();
        // 分页过滤器：PageFilter
        //查询前3页数据，每10条数据为一页
        xuQiu5();


    }

    public static void xuQiu5() throws Exception {
        //获取表对象
        Table students2 = HbaseTool.getOneTable("students2");
        Scan scan = new Scan();

        //定义页数
        int pageNum = 3;
        //定义每页的条数
        int pageDataNum = 10;
        //定义开始行
        String startRowKey = "";

        //创建一个分页过滤器
        PageFilter pageFilter = new PageFilter(pageDataNum);

        for (int i = 1; i <= pageNum; i++) {
            System.out.println("=======================当前是第 " + i + " 页数据=======================");
            scan.withStartRow(Bytes.toBytes(startRowKey));
            scan.setFilter(pageFilter);
            ResultScanner scanner = students2.getScanner(scan);
            for (Result result : scanner) {
                HbaseTool.printResult(result);
                startRowKey = Bytes.toString(result.getRow()) + "0";
            }
        }


        HbaseTool.close();

    }

    public static void xuQiu4() throws Exception {
        //获取表对象
        Table students2 = HbaseTool.getOneTable("students2");
        Scan scan = new Scan();

        PrefixFilter prefixFilter = new PrefixFilter(Bytes.toBytes("15001005"));

        //设置过滤器
        //Scan setFilter(Filter filter)
        scan.setFilter(prefixFilter);

        ResultScanner studentsScanner = students2.getScanner(scan);
        for (Result rs : studentsScanner) {
            HbaseTool.printResult(rs);
        }

        HbaseTool.close();
    }

    public static void xuQiu3() throws Exception {
        //获取表对象
        Table students2 = HbaseTool.getOneTable("students2");
        Scan scan = new Scan();


        RowFilter rowFilter = new RowFilter(CompareOperator.EQUAL, new BinaryPrefixComparator(Bytes.toBytes("15001005")));

        //设置过滤器
        //Scan setFilter(Filter filter)
        scan.setFilter(rowFilter);

        ResultScanner studentsScanner = students2.getScanner(scan);
        for (Result rs : studentsScanner) {
            HbaseTool.printResult(rs);
        }

        HbaseTool.close();
    }


    public static void xuQiu2() throws Exception {
        //获取表对象
        Table students2 = HbaseTool.getOneTable("students2");
        Scan scan = new Scan();

        //单列值排除过滤器
        //SingleColumnValueExcludeFilter(final byte[] family, final byte[] qualifier,
        //      final CompareOp compareOp, ByteArrayComparable comparator)
        SingleColumnValueExcludeFilter singleColumnValueExcludeFilter = new SingleColumnValueExcludeFilter(Bytes.toBytes("info"), Bytes.toBytes("name"),
                CompareOperator.EQUAL, new BinaryPrefixComparator(Bytes.toBytes("张")));

        //设置过滤器
        //Scan setFilter(Filter filter)
        scan.setFilter(singleColumnValueExcludeFilter);

        ResultScanner studentsScanner = students2.getScanner(scan);
        for (Result rs : studentsScanner) {
            HbaseTool.printResult(rs);
        }

        HbaseTool.close();
    }

    public static void xuQiu1() throws Exception {
        //获取表对象
        Table students2 = HbaseTool.getOneTable("students2");
        Scan scan = new Scan();

        //SingleColumnValueFilter
        //SingleColumnValueFilter(final byte [] family, final byte [] qualifier,
        //      final CompareOperator op,
        //      final org.apache.hadoop.hbase.filter.ByteArrayComparable comparator)
        SingleColumnValueFilter singleColumnValueFilter = new SingleColumnValueFilter(Bytes.toBytes("info"), Bytes.toBytes("name"),
                CompareOperator.EQUAL, new BinaryPrefixComparator(Bytes.toBytes("张")));

        //设置过滤器
        //Scan setFilter(Filter filter)
        scan.setFilter(singleColumnValueFilter);

        ResultScanner studentsScanner = students2.getScanner(scan);
        for (Result rs : studentsScanner) {
            HbaseTool.printResult(rs);
        }

        HbaseTool.close();
    }


}
